import threading

import redis

from model.db.name_mapping import refresh_name_cache
from utils.utils import redis_client, REDIS_STREAM_KEY, GROUP_NAME, CONSUMER_NAME


def publish_refresh_event_streams(name_type):
    message = {
        'action': 'refresh',
        'name_type': name_type
    }
    redis_client.xadd(REDIS_STREAM_KEY, message)
    print(f"[REDIS STREAM] Published refresh event for name_type={name_type}")


def create_group():
    try:
        redis_client.xgroup_create(REDIS_STREAM_KEY, GROUP_NAME, id='0', mkstream=True)
        print(f"[INIT] Consumer group '{GROUP_NAME}' created.")
    except redis.exceptions.ResponseError as e:
        if "BUSYGROUP" in str(e):
            print(f"[INIT] Consumer group '{GROUP_NAME}' already exists.")
        else:
            raise


def start_consumer_redis_stream(app):
    '''
    子线程去消费streams中的消息并做业务处理
    '''

    create_group()

    print(f"[STREAM CONSUMER] Start consuming from stream: {REDIS_STREAM_KEY}")

    def _consume_stream_msg():
        while True:
            try:
                response = redis_client.xreadgroup(
                    groupname=GROUP_NAME,
                    consumername=CONSUMER_NAME,
                    streams={REDIS_STREAM_KEY: '>'},
                    count=10,
                    block=5000  # 5秒阻塞
                )

                if response:
                    for stream_key, messages in response:
                        for message_id, message_data in messages:
                            action = message_data.get(b'action')
                            name_type = message_data.get(b'name_type')

                            if action and action.decode() == 'refresh':
                                print(f"[STREAM CONSUMER] Received refresh for name_type={name_type.decode()}")

                                # 刷新缓存
                                with app.app_context():
                                    refresh_name_cache()

                                # 消费确认
                                redis_client.xack(REDIS_STREAM_KEY, GROUP_NAME, message_id)

            except Exception as e:
                print("[ERROR] Consumer error:", e)

    # 启动消费线程
    thread = threading.Thread(target=_consume_stream_msg)
    # Notice 设置为守护线程 跟随flask项目主线程共同停止
    thread.daemon = True
    thread.start()
